package cn.fetosoft.woodpecker.core.pubsub.zookeeper;

import cn.fetosoft.woodpecker.core.pubsub.Subscription;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * @author guobingbing
 * @wechat t_gbinb
 * @since 2021/9/12 11:07
 */
public abstract class AbstractZkSubscription extends AbstractZookeeper implements Subscription {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractZkSubscription.class);

    @Override
    public void subscript() throws Exception {
        String path = this.getSubPath();
        if(StringUtils.isNotBlank(path)) {
            CuratorCache cache = CuratorCache.build(client, path);
            CuratorCacheListener listener = CuratorCacheListener.builder()
                    .forChanges(AbstractZkSubscription.this::subNodeChanged)
                    .build();
            cache.listenable().addListener(listener);
            cache.start();
            this.createNotExist(path, CreateMode.PERSISTENT);
        }else{
            LOGGER.info("The path is null or empty!");
        }
    }

    /**
     * 查询所有注册的节点
     * @return
     * @throws Exception
     */
    @Override
    public List<NodeEntity> getRegisterClusters() throws Exception{
        List<NodeEntity> nodes = new ArrayList<>();
        String path = this.getRegPath();
        if(StringUtils.isNotBlank(path)){
            List<String> children = this.client.getChildren().forPath(path);
            for(String child : children){
                byte[] b = this.client.getData().forPath(path + "/" + child);
                String data = new String(b, CHARSET);
                NodeEntity entity = JSON.toJavaObject(JSONObject.parseObject(data), NodeEntity.class);
                nodes.add(entity);
            }
        }
        return nodes;
    }

    /**
     * 获取订阅的Path
     * @author guobingbing
     * @date 2020-03-09 13:12
     * @param
     * @return ZkClientBuilder
     */
    protected abstract String getSubPath();

    /**
     * 节点注册地址
     * @return
     */
    protected abstract String getRegPath();

    /**
     * 订阅节点变动
     * @author guobingbing
     * @date 2020-03-09 14:03
     * @param oldData
     * @param data
     * @return void
     */
    protected abstract void subNodeChanged(ChildData oldData, ChildData data);
}
